import org.apache.spark.SparkConf;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.streaming.Durations;
import org.apache.spark.streaming.api.java.JavaDStream;
import org.apache.spark.streaming.api.java.JavaPairDStream;
import org.apache.spark.streaming.api.java.JavaReceiverInputDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;
import scala.Tuple2;

import java.util.Arrays;
import java.util.Iterator;

/**
 * @author zxj<br>
 * 时间 2018/1/31 15:06
 * 说明 ...
 */
public class StreamingDemo {

    public static void main(String[] args) throws InterruptedException {
        //如果设置为local运行，则必须设置2个或者以上的线程，因为spark-streaming运行的过程中，至少需要一个线程不断的接收数据，另一个线程消费数据
        //否则有可能只接收不消费数据
        SparkConf conf = new SparkConf().setMaster("local[2]").setAppName("word");
        //我们把一个批次的时间设置为2秒
        JavaStreamingContext jssc = new JavaStreamingContext(conf, Durations.seconds(2));
        //这里设置数据来源，输入来源可以是任何机器的
        //这里是消费其它机器的一个SOCKET数据，该机器需要启动一个SOCKET SERVER，端口是9999
        JavaReceiverInputDStream<String> lines = jssc.socketTextStream("192.18.16.222",9999);


        //对单词进行切割
        JavaDStream<String>  words = lines.flatMap(new FlatMapFunction<String, String>() {

            @Override
            public Iterator<String> call(String s) throws Exception {
                return Arrays.asList(s.split(" ")).iterator();
            }
        });

        //转换成pair格式，一个字符一个次数
        JavaPairDStream<String, Integer> pairs = words.mapToPair(new PairFunction<String, String, Integer>() {

            @Override
            public Tuple2<String, Integer> call(String word) throws Exception {
                return new Tuple2<String, Integer>(word, 1);
            }
        });

        //reduce
        JavaPairDStream<String, Integer> count = pairs.reduceByKey(new Function2<Integer, Integer, Integer>() {
            @Override
            public Integer call(Integer x, Integer y) throws Exception {
                return x + y;
            }
        });
        //输出结果，这个函数只会输出前面十个
        count.print();

        JavaPairDStream<String, Integer> window = count.window(Durations.seconds(20), Durations.seconds(10));
        window.print();
        System.out.println("window over");

        jssc.start();
        jssc.awaitTermination();

    }
}
